package me.sdau.spider.kafkaprocess.service;

import lombok.extern.slf4j.Slf4j;
import me.sdau.spider.kafkaprocess.comm.utils.FileUtils;
import me.sdau.spider.kafkaprocess.kafka.dto.ConfigDTO;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;

/**
 * @author zj
 * @date 2018/4/3 16:16
 */

@Slf4j
@Service
public class ElasticSearchService {
    private static TransportClient transportClient;

    @Autowired
    private void setTransportClient(TransportClient transportClient) {
        ElasticSearchService.transportClient = transportClient;
    }

    public void consumData(String key, String value) {
        List<ConfigDTO> index = Optional.ofNullable(FileUtils.getIndex())
                .orElse(new ArrayList<>());
        System.out.println("key:" + key + ",val:" + value);
        index.stream()
                .filter(c -> Objects.equals(c.getTopic(), key))
                .forEach(p -> transportClient.prepareIndex(p.getIndex().toLowerCase(), key)
                        .setSource(value, XContentType.JSON)
                        .get());
    }

    public void batchConsum(List<ConsumerRecord<String, String>> records) {
        List<ConfigDTO> index = Optional.ofNullable(FileUtils.getIndex())
                .orElse(new ArrayList<>());

        BulkRequestBuilder prepareBulk = transportClient.prepareBulk();
        records.forEach(record -> index.stream()
                    .filter(c -> Objects.equals(c.getTopic(), record.key()))
                    .forEach(p -> prepareBulk.add(transportClient
                            .prepareIndex(p.getIndex().toLowerCase(), record.key())
                            .setSource(record.value(), XContentType.JSON))));
        prepareBulk.get();
    }


}
